// DemandPredictionMapper.java
package com.taxi.analysis.demand;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 实时需求预测Mapper
 * 功能：处理出租车需求数据，生成多维度分区数据
 * 技术：分区技术
 */
public class DemandPredictionMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 跳过CSV标题行
        if (key.get() == 0) {
            return;
        }

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) {
                return;
            }

            // 解析CSV数据行
            String[] fields = parseCSVLine(line);

            // CSV字段索引:
            // 0:tpep_pickup_datetime, 1:PULocationID, 2:DOLocationID,
            // 3:pickup_year, 4:pickup_month, 5:pickup_day, 6:pickup_hour, 7:pickup_minute,
            // 8:pickup_weekday, 9:pickup_week, 10:time_period, 11:day_type, 12:season,
            // 13:trip_distance, 14:total_amount, 15:trip_duration, 16:avg_speed, 17:passenger_count,
            // 18:pickup_density, 19:dropoff_density, 20:fare_level, 21:distance_level,
            // 22:time_partition, 23:geo_partition, 24:demand_partition, 25:composite_partition,
            // 26:hourly_demand_count, 27:period_demand_count, 28:daytype_demand_count,
            // 29:demand_intensity, 30:demand_level, 31:demand_trend

            if (fields.length >= 32) {
                String pickupDatetime = fields[0].trim();
                String pickupLocationId = fields[1].trim();
                String dropoffLocationId = fields[2].trim();
                int pickupYear = parseInt(fields[3]);
                int pickupMonth = parseInt(fields[4]);
                int pickupDay = parseInt(fields[5]);
                int pickupHour = parseInt(fields[6]);
                int pickupMinute = parseInt(fields[7]);
                int pickupWeekday = parseInt(fields[8]);
                int pickupWeek = parseInt(fields[9]);
                String timePeriod = fields[10].trim();
                String dayType = fields[11].trim();
                String season = fields[12].trim();
                double tripDistance = parseDouble(fields[13]);
                double totalAmount = parseDouble(fields[14]);
                double tripDuration = parseDouble(fields[15]);
                double avgSpeed = parseDouble(fields[16]);
                int passengerCount = parseInt(fields[17]);
                String pickupDensity = fields[18].trim();
                String dropoffDensity = fields[19].trim();
                String fareLevel = fields[20].trim();
                String distanceLevel = fields[21].trim();
                String timePartition = fields[22].trim();
                String geoPartition = fields[23].trim();
                String demandPartition = fields[24].trim();
                String compositePartition = fields[25].trim();
                int hourlyDemandCount = parseInt(fields[26]);
                int periodDemandCount = parseInt(fields[27]);
                int daytypeDemandCount = parseInt(fields[28]);
                double demandIntensity = parseDouble(fields[29]);
                String demandLevel = fields[30].trim();
                String demandTrend = fields[31].trim();

                // 验证数据有效性
                if (isValidData(pickupLocationId, dropoffLocationId, pickupHour)) {

                    // 生成分区数据1: 综合分区（时间+地理+需求类型）
                    generateCompositePartition(compositePartition, pickupLocationId, dropoffLocationId,
                            pickupDatetime, pickupHour, pickupMinute, timePeriod, dayType,
                            tripDistance, totalAmount, tripDuration, passengerCount,
                            pickupDensity, fareLevel, distanceLevel, demandIntensity,
                            demandLevel, demandTrend, context);

                    // 生成分区数据2: 时间分区（按小时分区）
                    generateTimePartition(timePartition, pickupLocationId, pickupDatetime,
                            timePeriod, dayType, season, tripDistance, totalAmount,
                            demandIntensity, demandLevel, hourlyDemandCount, context);

                    // 生成分区数据3: 地理分区（按地点分区）
                    generateGeoPartition(geoPartition, pickupLocationId, dropoffLocationId,
                            pickupHour, timePeriod, dayType, tripDistance, totalAmount,
                            pickupDensity, dropoffDensity, demandIntensity, demandLevel, context);

                    // 生成分区数据4: 需求类型分区（按需求特征分区）
                    generateDemandTypePartition(demandPartition, pickupLocationId, pickupHour,
                            timePeriod, fareLevel, distanceLevel, demandLevel,
                            demandTrend, demandIntensity, totalAmount, tripDistance, context);
                }
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
            System.err.println("错误信息: " + e.getMessage());
        }
    }

    /**
     * 生成综合分区数据
     */
    private void generateCompositePartition(String compositePartition, String pickupLocationId,
                                            String dropoffLocationId, String pickupDatetime, int pickupHour,
                                            int pickupMinute, String timePeriod, String dayType,
                                            double tripDistance, double totalAmount, double tripDuration,
                                            int passengerCount, String pickupDensity, String fareLevel,
                                            String distanceLevel, double demandIntensity, String demandLevel,
                                            String demandTrend, Context context)
            throws IOException, InterruptedException {

        String partitionKey = "PARTITION_" + compositePartition;
        String demandRecord = createDemandRecord(pickupLocationId, dropoffLocationId, pickupDatetime,
                pickupHour, pickupMinute, timePeriod, dayType, tripDistance,
                totalAmount, tripDuration, passengerCount, pickupDensity,
                fareLevel, distanceLevel, demandIntensity, demandLevel, demandTrend);

        outputKey.set(partitionKey);
        outputValue.set("COMPOSITE_DEMAND|" + demandRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 生成时间分区数据
     */
    private void generateTimePartition(String timePartition, String pickupLocationId, String pickupDatetime,
                                       String timePeriod, String dayType, String season, double tripDistance,
                                       double totalAmount, double demandIntensity, String demandLevel,
                                       int hourlyDemandCount, Context context)
            throws IOException, InterruptedException {

        String partitionKey = "TIME_" + timePartition;
        String timeRecord = createTimeRecord(pickupLocationId, pickupDatetime, timePeriod, dayType,
                season, tripDistance, totalAmount, demandIntensity,
                demandLevel, hourlyDemandCount);

        outputKey.set(partitionKey);
        outputValue.set("TIME_DEMAND|" + timeRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 生成地理分区数据
     */
    private void generateGeoPartition(String geoPartition, String pickupLocationId, String dropoffLocationId,
                                      int pickupHour, String timePeriod, String dayType, double tripDistance,
                                      double totalAmount, String pickupDensity, String dropoffDensity,
                                      double demandIntensity, String demandLevel, Context context)
            throws IOException, InterruptedException {

        String partitionKey = "GEO_" + geoPartition;
        String geoRecord = createGeoRecord(pickupLocationId, dropoffLocationId, pickupHour, timePeriod,
                dayType, tripDistance, totalAmount, pickupDensity, dropoffDensity,
                demandIntensity, demandLevel);

        outputKey.set(partitionKey);
        outputValue.set("GEO_DEMAND|" + geoRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 生成需求类型分区数据
     */
    private void generateDemandTypePartition(String demandPartition, String pickupLocationId, int pickupHour,
                                             String timePeriod, String fareLevel, String distanceLevel,
                                             String demandLevel, String demandTrend, double demandIntensity,
                                             double totalAmount, double tripDistance, Context context)
            throws IOException, InterruptedException {

        String partitionKey = "DEMAND_" + demandPartition;
        String demandTypeRecord = createDemandTypeRecord(pickupLocationId, pickupHour, timePeriod,
                fareLevel, distanceLevel, demandLevel, demandTrend,
                demandIntensity, totalAmount, tripDistance);

        outputKey.set(partitionKey);
        outputValue.set("DEMAND_TYPE|" + demandTypeRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 创建需求记录
     */
    private String createDemandRecord(String pickupLocationId, String dropoffLocationId, String pickupDatetime,
                                      int pickupHour, int pickupMinute, String timePeriod, String dayType,
                                      double tripDistance, double totalAmount, double tripDuration,
                                      int passengerCount, String pickupDensity, String fareLevel,
                                      String distanceLevel, double demandIntensity, String demandLevel,
                                      String demandTrend) {

        return String.format("%s|%s|%s|%d|%d|%s|%s|%.2f|%.2f|%.2f|%d|%s|%s|%s|%.2f|%s|%s",
                pickupLocationId, dropoffLocationId, pickupDatetime, pickupHour, pickupMinute,
                timePeriod, dayType, tripDistance, totalAmount, tripDuration, passengerCount,
                pickupDensity, fareLevel, distanceLevel, demandIntensity, demandLevel, demandTrend);
    }

    /**
     * 创建时间记录
     */
    private String createTimeRecord(String pickupLocationId, String pickupDatetime, String timePeriod,
                                    String dayType, String season, double tripDistance, double totalAmount,
                                    double demandIntensity, String demandLevel, int hourlyDemandCount) {

        return String.format("%s|%s|%s|%s|%s|%.2f|%.2f|%.2f|%s|%d",
                pickupLocationId, pickupDatetime, timePeriod, dayType, season,
                tripDistance, totalAmount, demandIntensity, demandLevel, hourlyDemandCount);
    }

    /**
     * 创建地理记录
     */
    private String createGeoRecord(String pickupLocationId, String dropoffLocationId, int pickupHour,
                                   String timePeriod, String dayType, double tripDistance, double totalAmount,
                                   String pickupDensity, String dropoffDensity, double demandIntensity,
                                   String demandLevel) {

        return String.format("%s|%s|%d|%s|%s|%.2f|%.2f|%s|%s|%.2f|%s",
                pickupLocationId, dropoffLocationId, pickupHour, timePeriod, dayType,
                tripDistance, totalAmount, pickupDensity, dropoffDensity, demandIntensity, demandLevel);
    }

    /**
     * 创建需求类型记录
     */
    private String createDemandTypeRecord(String pickupLocationId, int pickupHour, String timePeriod,
                                          String fareLevel, String distanceLevel, String demandLevel,
                                          String demandTrend, double demandIntensity, double totalAmount,
                                          double tripDistance) {

        return String.format("%s|%d|%s|%s|%s|%s|%s|%.2f|%.2f|%.2f",
                pickupLocationId, pickupHour, timePeriod, fareLevel, distanceLevel,
                demandLevel, demandTrend, demandIntensity, totalAmount, tripDistance);
    }

    /**
     * 验证数据有效性
     */
    private boolean isValidData(String pickupLocationId, String dropoffLocationId, int pickupHour) {
        return !pickupLocationId.isEmpty() && !dropoffLocationId.isEmpty() &&
                !pickupLocationId.equals("null") && !dropoffLocationId.equals("null") &&
                pickupHour >= 0 && pickupHour <= 23;
    }

    /**
     * 解析CSV行
     */
    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    /**
     * 处理包含引号的CSV行
     */
    private String[] parseQuotedCSV(String line) {
        java.util.List<String> fields = new java.util.ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    /**
     * 安全的double解析
     */
    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    /**
     * 安全的int解析
     */
    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
